/*
 * @Author: caiwanli 651943559@qq.com
 * @Date: 2023-12-19 16:12:34
 * @LastEditors: caiwanli 651943559@qq.com
 * @LastEditTime: 2024-01-18 13:12:46
 * @FilePath: /task_sche_parser/src/planner/plan_select.cpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置
 * 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */

#include "abstract_expression.h"
#include "abstract_plan.h"
#include "bound_alias.h"
#include "bound_base_table_ref.h"
#include "bound_binary_op.h"
#include "bound_cross_product_ref.h"
#include "bound_expression.h"
#include "bound_expression_list_ref.h"
#include "bound_join_ref.h"
#include "bound_table_ref.h"
#include "column_value_expression.h"
#include "compare_expression.h"
#include "constant_value_expression.h"
#include "exception.h"
#include "logic_expression.h"
#include "filter_plan.h"
#include "logical_type.hpp"
#include "nest_loop_join_plan.h"
#include "physical_operator.hpp"
#include "planner.hpp"
#include "projection_plan.h"
#include "seq_scan_plan.h"
#include "agg_plan.h"
#include <array>
#include <cstdio>

namespace DaseX {

auto Planner::PlanSelect(const SelectStatement &statement)
    -> AbstractPlanNodeRef {

  AbstractPlanNodeRef plan = nullptr;

  printf("planSelect start\n");

  switch (statement.table_->type_) {
  // tableref between from - where
  case TableReferenceType::BASE_TABLE:
  case TableReferenceType::CROSS_PRODUCT:
  case TableReferenceType::JOIN:
    plan = PlanTableRef(*statement.table_);
  }

  printf("from finish \n");

  // plan where, to construct filter node
  if (!statement.where_->IsInvalid()) {
    auto schema = plan->OutputSchema();
    // just for debug
    // to del
    switch (statement.where_->type_) {
    case ExpressionType::BINARY_OP: {
      printf("binary\n");
      break;
    }
    case ExpressionType::COLUMN_REF: {
      printf(" col \n");
      break;
    }
    }
    auto temp_tuple = PlanExpression(*statement.where_, {plan});
    auto expr = std::get<1>(temp_tuple);

    printf("todo : filter,expr type = \n");
    // why tuple error?  0311:work at agg try?
    // auto [_, expr] = PlanExpression(*statement.where_, {plan});
    plan = std::make_shared<FilterPlanNode>(
        std::make_shared<arrow::Schema>(schema), std::move(expr),
        std::move(plan));
  }

  printf("where finish\n");

  bool has_agg = false;
  for (const auto &item : statement.select_list_) {
    if (item->HasAggregation()) {
      has_agg = true;
      break;
    }
  }

  if (!statement.having_->IsInvalid() || !statement.group_by_.empty() ||
      has_agg) {
    // TODO process agg
    plan = PlanSelectAgg(statement, std::move(plan));
  } else {
    // 普通select
    std::vector<AbstractExpressionRef> exprs;
    std::vector<std::string> column_names;
    std::vector<AbstractPlanNodeRef> children = {plan};
    for (const auto &item : statement.select_list_) {
      auto [name, expr] = PlanExpression(*item, {plan});
      exprs.emplace_back(std::move(expr));
      column_names.emplace_back(std::move(name));
    }
    printf("proj\n");
    plan = std::make_shared<ProjectionPlanNode>(
        std::make_shared<arrow::Schema>(ProjectionPlanNode::RenameSchema(
            ProjectionPlanNode::InferProjectionSchema(exprs), column_names)),
        std::move(exprs), std::move(plan));
  }

  // 处理orderby
  // if (!statement.sort_.empty()) {
  //   for (const auto &order_by : statement.sort_) {
  //     // todo
  //   }
  // }
  return plan;
}

auto Planner::PlanTableRef(const BoundTableRef &table_ref)
    -> AbstractPlanNodeRef {
  printf("planTableRef\n");
  switch (table_ref.type_) {
  case TableReferenceType::BASE_TABLE: {
    const auto &base_table_ref =
        dynamic_cast<const BoundBaseTableRef &>(table_ref);
    return PlanBaseTableRef(base_table_ref);
  }
  case TableReferenceType::CROSS_PRODUCT: {
    const auto &cross_product =
        dynamic_cast<const BoundCrossProductRef &>(table_ref);
    return PlanCrossProductRef(cross_product);
  }
  case TableReferenceType::JOIN: {
    const auto &join = dynamic_cast<const BoundJoinRef &>(table_ref);
    printf("join weak support now\n");
    return PlanJoinRef(join);
  }
  case TableReferenceType::EXPRESSION_LIST: {
    const auto &expression_list =
        dynamic_cast<const BoundExpressionListRef &>(table_ref);
    // return PlanExpressionListRef(expression_list);
    printf("expressonlist not support now\n");
  }
  case TableReferenceType::SUBQUERY: {
    const auto &subquery = dynamic_cast<const BoundSubqueryRef &>(table_ref);
    // return PlanSubquery(subquery, subquery.alias_);
    printf("SUBQUERY not support now\n");
  }

  default:
    break;
  }
  throw DaseX::Exception(
      "the table ref type {} is not supported in planner yet");
}

auto Planner::PlanBaseTableRef(const BoundBaseTableRef &table_ref)
    -> AbstractPlanNodeRef {
  //
  std::shared_ptr<Table> table;
  auto RC = catalog_.get_table(table_ref.table_, table);

  // plan as normal SeqScan.
  printf("seq scan plan node\n");
  // TODO:fix plan node.
  // 目前我直接把schema返回去了，对多表应该返回一个x.y列名的临时元数据表的
  // inferscanschema 中 finish todo return std::make_shared<SeqScanPlanNode>(
  // table->schema,table->table_name);
  return std::make_shared<SeqScanPlanNode>(
      std::make_shared<arrow::Schema>(
          SeqScanPlanNode::InferScanSchema(table_ref)),
      table->table_name);
}

auto Planner::PlanCrossProductRef(const BoundCrossProductRef &table_ref)
    -> AbstractPlanNodeRef {
  auto left = PlanTableRef(*table_ref.left_);
  auto right = PlanTableRef(*table_ref.right_);

  // check
  printf("%d\n", left->GetType());
  printf("%d\n", right->GetType());
  printf("笛卡尔积\n");
  // crossproduct
  std::shared_ptr<Table> table;

  return std::make_shared<NestedLoopJoinPlanNode>(
      std::make_shared<arrow::Schema>(
          NestedLoopJoinPlanNode::InferJoinSchema(*left, *right)),
      std::move(left), std::move(right),
      std::make_shared<ConstantValueExpression>(1), JoinType::INNER);
}

auto Planner::PlanColumnRef(const BoundColumnRef &expr,
                            const std::vector<AbstractPlanNodeRef> &children)
    -> std::tuple<std::string, std::shared_ptr<ColumnValueExpression>> {
  if (children.empty()) {
    throw Exception("column ref should have at least one child");
  }

  auto col_name = expr.ToString();
  printf("expr name :%s\n", col_name.c_str());

  if (children.size() == 1) {
    // Projections, Filters, and other executors evaluating expressions with one
    // single child will use this branch.
    printf("Projections, Filters,  expressions with one single child will use "
           "this branch.\n");
    const auto &child = children[0];
    auto schema = child->OutputSchema();
    uint32_t col_idx = schema.GetFieldIndex(col_name);
    // auto col_type = schema.GetColumn(col_idx).GetType();
    auto col_type = LogicalType::INTEGER;
    return std::make_tuple(col_name, std::make_shared<ColumnValueExpression>(
                                         0, col_idx, col_type));
  }
  if (children.size() == 2) {
    /*
     * Joins will use this branch to plan expressions.
     */
    printf("Joins will use this branch to plan expressions.\n");
    const auto &left = children[0];
    const auto &right = children[1];
    auto left_schema = left->OutputSchema();
    auto right_schema = right->OutputSchema();

    auto col_idx_left = left_schema.GetFieldIndex(col_name);
    auto col_idx_right = right_schema.GetFieldIndex(col_name);
    if (col_idx_left != -1) {
      // auto col_type = left_schema.GetColumn(*col_idx_left).GetType();
      auto col_type = LogicalType::INTEGER;
      return std::make_tuple(col_name, std::make_shared<ColumnValueExpression>(
                                           0, col_idx_left, col_type));
    }
    if (col_idx_right != -1) {
      // auto col_type = right_schema.GetColumn(*col_idx_right).GetType();
      auto col_type = LogicalType::INTEGER;
      return std::make_tuple(col_name, std::make_shared<ColumnValueExpression>(
                                           1, col_idx_right, col_type));
    }
  }
}

auto Planner::PlanBinaryOp(const BoundBinaryOp &expr,
                           const std::vector<AbstractPlanNodeRef> &children)
    -> AbstractExpressionRef {
  auto [_1, left] = PlanExpression(*expr.larg_, children);
  auto [_2, right] = PlanExpression(*expr.rarg_, children);
  const auto &op_name = expr.op_name_;
  if (op_name == "=" || op_name == "==") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::Equal);
  }
  if (op_name == "!=") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::NotEqual);
  }
  if (op_name == "<") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::LessThan);
  }
  if (op_name == "<=") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::LessThanOrEqual);
  }
  if (op_name == ">") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::GreaterThan);
  }
  if (op_name == ">=") {
    return std::make_shared<ComparisonExpression>(
        std::move(left), std::move(right), ComparisonType::GreaterThanOrEqual);
  }
  if (op_name == "and") {
    return std::make_shared<LogicExpression>(std::move(left), std::move(right), LogicType::And);
  }

  // return GetBinaryExpression(op_name, std::move(left), std::move(right));
}

auto Planner::PlanConstant(const BoundConstant &expr,
                           const std::vector<AbstractPlanNodeRef> &children)
    -> AbstractExpressionRef {
  return std::make_shared<ConstantValueExpression>(expr.val_);
}

auto Planner::PlanExpression(const BoundExpression &expr,
                             const std::vector<AbstractPlanNodeRef> &children)
    -> std::tuple<std::string, AbstractExpressionRef> {
  switch (expr.type_) {
  // case ExpressionType::AGG_CALL: {
  //   return std::make_tuple("UNNAMED_COLUMN",
  //   std::move(ctx_.expr_in_agg_[ctx_.next_aggregation_++]));
  // }
  case ExpressionType::COLUMN_REF: {
    const auto &column_ref_expr = dynamic_cast<const BoundColumnRef &>(expr);
    return PlanColumnRef(column_ref_expr, children);
  }
  case ExpressionType::BINARY_OP: {
    const auto &binary_op_expr = dynamic_cast<const BoundBinaryOp &>(expr);
    return std::make_tuple("unnamed", PlanBinaryOp(binary_op_expr, children));
  }
  case ExpressionType::CONSTANT: {
    const auto &constant_expr = dynamic_cast<const BoundConstant &>(expr);
    return std::make_tuple("unnamed const",
                           PlanConstant(constant_expr, children));
  }
  default:
    break;
  }
  printf("%d", expr.type_);
  throw Exception("expression type {} not supported in planner yet,because "
                  "PlanColumnRef not finish");
}

auto Planner::PlanJoinRef(const BoundJoinRef &table_ref)
    -> AbstractPlanNodeRef {
  auto left = PlanTableRef(*table_ref.left_);
  auto right = PlanTableRef(*table_ref.right_);
  std::vector<AbstractPlanNodeRef> children;
  children.push_back(left);
  children.push_back(right);
  auto temp_tuple = PlanExpression(*table_ref.condition_, children);
  auto [_, join_condition] =
      PlanExpression(*table_ref.condition_, {left, right});
  auto nlj_node = std::make_shared<NestedLoopJoinPlanNode>(
      std::make_shared<arrow::Schema>(
          NestedLoopJoinPlanNode::InferJoinSchema(*left, *right)),
      std::move(left), std::move(right), std::move(join_condition),
      table_ref.join_type_);
  return nlj_node;
}

auto Planner::PlanSelectAgg(const SelectStatement &statement,
                            AbstractPlanNodeRef child) -> AbstractPlanNodeRef {
  // TODO(zyy): agg idx in outputschema

  std::vector<AbstractExpressionRef> group_by_expr;
  std::vector<std::string> outputschema_col_names;

  for(const auto &expr : statement.group_by_) {
    auto [col_name,ab_expr] = PlanExpression(*expr,{child});
    group_by_expr.emplace_back(std::move(ab_expr));
    outputschema_col_names.emplace_back(std::move(col_name));
  }







  return DaseX::AbstractPlanNodeRef();
}

} // namespace DaseX